package keter.ssh.exec;

import java.io.File;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;

import org.apache.commons.lang3.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Lists;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;

import keter.framework.util.IOUtil;
import keter.ssh.core.SshUtil;

public class CmdExecutor {
	
	private static final Logger logger = LoggerFactory.getLogger(CmdExecutor.class);
	
	private static ListeningExecutorService pool;
	
	private static List<ListenableFuture<String>> futures;
	
	private static int thread=1;
	private static boolean sequence=false;
	
	public static void init(int thread){
		CmdExecutor.thread = thread;
	}
	
	public  static void execute(List<SshUtil> sshList, String cmd){
		Validate.notNull(sshList,"ssh清单不能为空！"); 
		pool = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(thread));
		futures = Lists.newArrayList();
		for (SshUtil ssh : sshList) {
			if(ssh!=null){
				ListenableFuture<String> future = pool.submit(new Callable<String>() {
					@Override
					public String call() throws Exception {
						return exec(cmd, ssh);
					}
				});
				futures.add(future);
			}
		}
		execFutures(futures);
	}
	
	private static String exec(String cmd, SshUtil ssh){
		try {
			return "\n--------  "+ssh.getHost().getAddr()+"  ------------\n"
		           +ssh.exec(cmd).result()
		           +"\n--------------------------------------\n";
		} catch (Exception e) {
			logger.error("主机:"+ssh.getHost().getAddr()+"命令执行失败!",e);
		}
		return "failure";
	}

	/**
	 * 执行futures
	 * @param future 
	 * @author gulixing@msn.com
	 * @date   2016年8月1日
	 */
	private static void execFutures(List<ListenableFuture<String>> futures) {
		//命令按顺序返回结果
		if(sequence){
			ListenableFuture<List<String>> successfulQueries = Futures.successfulAsList(futures);
			Futures.addCallback(successfulQueries, new FutureCallback<List<String>>() {
				  @Override
				  public void onSuccess(List<String> results) {
					  for (String result : results) {
					      System.out.println(result);					
					  }
				  }

				  @Override
				  public void onFailure(Throwable t) {
				      logger.error("执行失败",t);
				  }
		    });
		}
		else{
			for (ListenableFuture<String> future : futures) {
				Futures.addCallback(future, new FutureCallback<String>(){
					  @Override
					  public void onSuccess(String  result) {
						    System.out.println(result);
					  }
	
					  @Override
					  public void onFailure(Throwable t) {
					      logger.error("执行失败",t);
					  }
				});
			}
		}
	}
	
	public static void download(List<SshUtil> sshList, String src){
		String dest=".";
		download(sshList, src, dest);
	}
	
	/**
	 * 将文件下载到当前java虚拟机执行目录下的各自主机目录内
	 * @param src
	 * @param dest
	 * @throws CmdException 
	 * @author gulixing@msn.com
	 * @date   2016年7月17日
	 */
	public static void download(List<SshUtil> sshList, String src,String dest){
		pool = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(thread));
 		futures = Lists.newArrayList();
		for (SshUtil ssh : sshList) {
				if(ssh!=null){
				ListenableFuture<String> future = pool.submit(new Callable<String>() {
					String dest_by_host;
				    @Override
				    public String call(){
				    	try {
				    	    dest_by_host = dest+"/"+ssh.getHost().getAddr();
							IOUtil.ensureExists(dest_by_host);
							logger.info("下载文件[{}] 到 [{}]目录...",src,dest_by_host);
							ssh.download(src, dest_by_host);
						} catch (Exception e) {
						    logger.error("主机:"+ssh.getHost().getAddr()+"命令执行失败!",e);
						}
				    	return "success";
				    }
				});
				
				Futures.addCallback(future, new FutureCallback<String>() {
					  @Override
					  public void onSuccess(String result) {
						logger.info("主机{}文件传输完成!",ssh.getHost().getAddr());
					  }

					  @Override
					  public void onFailure(Throwable t) {
					      logger.error("执行失败",t);
					  }
			    });
			}
		}
	}
		
	public static void upload(List<SshUtil> sshList, String src){
		String dest="";
		upload(sshList, src, dest);
	}
	
	public static void upload(List<SshUtil> sshList, String src,String dest){
		pool = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(thread));
 		futures = Lists.newArrayList();
		for (SshUtil ssh : sshList) {
				if(ssh!=null){
				ListenableFuture<String> future = pool.submit(new Callable<String>() {
				    @Override
				    public String call(){
				    	try {
				    		//提取扩展名，合成目标URI
				    		String name = new File(src).getName();
				    		if(!dest.equals("")){
					    		String dest_with_fileName = dest+"/"+name;
								logger.info("文件[{}]上传到 [{}]",src,dest_with_fileName);
								ssh.upload(src, dest_with_fileName);
				    		}
				    		else{
								logger.info("文件[{}]上传到home目录...",src,name);
								ssh.upload(src, name);
				    		}
						} catch (Exception e) {
						    logger.error("主机:"+ssh.getHost().getAddr()+"命令执行失败!",e);
						}
				    	return "success";
				    }
				});
				
				Futures.addCallback(future, new FutureCallback<Object>() {
					  @Override
					  public void onSuccess(Object result) {
					    	logger.info("主机{}文件传输完成!",ssh.getHost().getAddr());
					  }

					  @Override
					  public void onFailure(Throwable t) {
					      logger.error("执行失败",t);
					  }
			        });
			}
		}
	}
}